package com.lzcu.maoyan.mapreduce.mapreduce;

import com.lzcu.maoyan.mapreduce.mapreduce.mapper.MovieMonthCountMapper;
import com.lzcu.maoyan.mapreduce.mapreduce.mapper.MovieTypeMapper;
import com.lzcu.maoyan.mapreduce.mapreduce.reduce.MovieMonthCountReduce;
import com.lzcu.maoyan.mapreduce.mapreduce.reduce.MovieTypeReduce;
import com.lzcu.maoyan.mapreduce.mapreduce.unit.MovieMonthCountWritable;
import com.lzcu.maoyan.mapreduce.mapreduce.unit.MovieTypeWritable;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.apache.phoenix.mapreduce.PhoenixInputFormat;
import org.apache.phoenix.mapreduce.PhoenixOutputFormat;
import org.apache.phoenix.mapreduce.util.PhoenixMapReduceUtil;

/**
 * 按月分析电影票房和电影数量
 * @author WangLiang
 * @date `2024/5/8`
 */
public class MovieMonthCount extends Configured implements Tool {

    @Override
    public int run(String[] strings) throws Exception {
        //1.创建job
        Configuration conf = new Configuration();
        conf.addResource("phoenix/core-site.xml");
        conf.addResource("phoenix/hbase-site.xml");
        conf.addResource("phoenix/hdfs-site.xml");
        Job job = Job.getInstance(conf, "PhoenixStockApplication");
        job.setJarByClass(PhoenixMR.class);
        //2.设置输入输出格式
        job.setInputFormatClass(PhoenixInputFormat.class);
        job.setOutputFormatClass(PhoenixOutputFormat.class);
        //3.设置数据读入和写出路径
        String selectQuery = "SELECT T1.MOVIE_NAME_INFO AS MOVIE_NAME, MAX(T1.TOTAL_SESSIONS) AS MAXTOTAL, T2.RELEASEDATE AS RELEASEDATE,T2.TYPEMOVIE AS SOURCE_MOVIE_TYPE\n" +
                "FROM MAOYANMOVIE.MOVIE_INFO T1 \n" +
                "INNER JOIN MAOYANMOVIE.MOVIE_DETAILED T2 \n" +
                "ON T1.MOVIE_NAME_INFO=T2.MOVIENAME\n" +
                "GROUP BY T1.MOVIE_NAME_INFO,T2.RELEASEDATE,T2.TYPEMOVIE";
        PhoenixMapReduceUtil.setInput(job, MovieMonthCountWritable.class,
                "MAOYANMOVIE.MOVIE_INFO,MAOYANMOVIE.MOVIE_DETAILED", selectQuery);
        PhoenixMapReduceUtil.setOutput(job, "MAOYANMOVIE.MOVIE_MONTH_COUNT",
                "TIME_DATA,MOVIE_COUNT,TOTAL,MOVIE_NAME");
        //4.设置代码片段
        job.setMapperClass(MovieMonthCountMapper.class);
        job.setReducerClass(MovieMonthCountReduce.class);
        //5.设置Mapper和Reducer端的输出key,value类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(MovieMonthCountWritable.class);
        TableMapReduceUtil.addDependencyJars(job);
        //6.任务提交
        return job.waitForCompletion(true) ? 0 : 1;
    }

    public static void main(String[] args) throws Exception {
        ToolRunner.run(new MovieMonthCount(), args);
    }

}
